package com.dmxh.orderMQ;

import com.dmxh.DO.DeathDo;
import com.dmxh.mapper.DeathMapper;
import com.dmxh.pojo.Death;
import com.dmxh.service.OrderService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;

@Component
@RocketMQMessageListener(consumerGroup = "2311", topic = "topicdk2310")
public class consumerMQ implements RocketMQListener<String> {/*, RocketMQPushConsumerLifecycleListener*/

    @Autowired
    private OrderService orderService;

    @Autowired
    private DeathMapper deathMapper;

    @Override
    public void onMessage(String s) {
        System.out.println(s); //这是生产者发送的消息
        System.out.println("消费者  启动!!!!");
        boolean b = orderService.queryOrderStatusAndDelete();
    }

    /*手动ACK返回操作   当重写prepareStart的时候onMessage就会失效*/
//    @Override//增强RocketMQ
//    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
//        /*1.注册消费者*/
//        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
//            @Override
//            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
//                /*2.通过list下标获取消息*/
//                MessageExt messageExt = list.get(0);
//                /*3.获取重试次数*/
//                int reconsumeTimes = messageExt.getReconsumeTimes();
//                /*4.获取消息主体 将byte数组转成String类型*/
//                String body = messageExt.getBody().toString();
//                /*5.获取topic*/
//                String topic = messageExt.getTopic();
//                try {
////                    int i = 1 / 0;   重试业务
//                    orderService.queryOrderStatusAndDelete();/*根据订单状态删除过期的订单*/
//                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;/*消费成功返回success*/
//                } catch (Exception e) {
//                    e.printStackTrace();
//                    if (reconsumeTimes > 1) {
//                        System.out.println("当前重试次数为:" + reconsumeTimes + "(保存到自定义的死信表中)");
//
//                        /*创建死信表对象*/
//                        DeathDo deathDo = new DeathDo();
//                        deathDo.setBody(body);
//                        deathDo.setError(e.getMessage());
//                        deathDo.setExttime(new Date());
//                        deathDo.setTopic(topic);
//                        deathDo.setFlag(0);
//                        deathDo.setMagid(messageExt.getMsgId());
//
//                        Death death = new Death();
//                        BeanUtils.copyProperties(deathDo, death);
//
//                        deathMapper.insertSelective(death);
//                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;/*死信消息增加成功*/
//                    }
//                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;/*重新消费之后*/
//                }
//            }
//        });
//    }
}
